/*
 * contrib/pgrowlocks/pgrowlocks.c
 *
 * Copyright (c) 2005-2006	Tatsuo Ishii
 *
 * Permission to use, copy, modify, and distribute this software and
 * its documentation for any purpose, without fee, and without a
 * written agreement is hereby granted, provided that the above
 * copyright notice and this paragraph and the following two
 * paragraphs appear in all copies.
 *
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE TO ANY PARTY FOR DIRECT,
 * INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING
 * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
 * DOCUMENTATION, EVEN IF THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED
 * OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * THE AUTHOR SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS ON AN "AS
 * IS" BASIS, AND THE AUTHOR HAS NO OBLIGATIONS TO PROVIDE MAINTENANCE,
 * SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 */

#include "postgres.h"

#include "access/heapam.h"
#include "access/multixact.h"
#include "access/relscan.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/namespace.h"
#include "catalog/pg_am_d.h"
#include "catalog/pg_authid.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "storage/bufmgr.h"
#include "storage/procarray.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/varlena.h"

PG_MODULE_MAGIC;

PG_FUNCTION_INFO_V1(pgrowlocks);

/* ----------
 * pgrowlocks:
 * 返回被锁定行的 tids
 * ----------
 */

#define NCHARS 32

#define		Atnum_tid		0
#define		Atnum_xmax		1
#define		Atnum_ismulti	2
#define		Atnum_xids		3
#define		Atnum_modes		4
#define		Atnum_pids		5

Datum pgrowlocks(PG_FUNCTION_ARGS)
{
	text	   *fc_relname = PG_GETARG_TEXT_PP(0);
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
	AttInMetadata *fc_attinmeta;
	Relation	fc_rel;
	RangeVar   *fc_relrv;
	TableScanDesc fc_scan;
	HeapScanDesc fc_hscan;
	HeapTuple	fc_tuple;
	AclResult	fc_aclresult;
	char	  **fc_values;

	InitMaterializedSRF(fcinfo, 0);

	/* 访问表 */
	fc_relrv = makeRangeVarFromNameList(textToQualifiedNameList(fc_relname));
	fc_rel = relation_openrv(fc_relrv, AccessShareLock);

	if (fc_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("\"%s\" is a partitioned table",
						RelationGetRelationName(fc_rel)),
				 errdetail("Partitioned tables do not contain rows.")));
	else if (fc_rel->rd_rel->relkind != RELKIND_RELATION)
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("\"%s\" is not a table",
						RelationGetRelationName(fc_rel))));
	else if (fc_rel->rd_rel->relam != HEAP_TABLE_AM_OID)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("only heap AM is supported")));

	/*
	 * 检查权限：必须对表拥有 SELECT 权限或在
	 * pg_stat_scan_tables 中
	 */
	fc_aclresult = pg_class_aclcheck(RelationGetRelid(fc_rel), GetUserId(),
								  ACL_SELECT);
	if (fc_aclresult != ACLCHECK_OK)
		fc_aclresult = has_privs_of_role(GetUserId(), ROLE_PG_STAT_SCAN_TABLES) ? ACLCHECK_OK : ACLCHECK_NO_PRIV;

	if (fc_aclresult != ACLCHECK_OK)
		aclcheck_error(fc_aclresult, get_relkind_objtype(fc_rel->rd_rel->relkind),
					   RelationGetRelationName(fc_rel));

	/* 扫描关系 */
	fc_scan = table_beginscan(fc_rel, GetActiveSnapshot(), 0, NULL);
	fc_hscan = (HeapScanDesc) fc_scan;

	fc_attinmeta = TupleDescGetAttInMetadata(fc_rsinfo->setDesc);

	fc_values = (char **) palloc(fc_rsinfo->setDesc->natts * sizeof(char *));

	while ((fc_tuple = heap_getnext(fc_scan, ForwardScanDirection)) != NULL)
	{
		TM_Result	fc_htsu;
		TransactionId fc_xmax;
		uint16		fc_infomask;

		/* 必须持有一个缓冲区锁才能调用 HeapTupleSatisfiesUpdate */
		LockBuffer(fc_hscan->rs_cbuf, BUFFER_LOCK_SHARE);

		fc_htsu = HeapTupleSatisfiesUpdate(fc_tuple,
										GetCurrentCommandId(false),
										fc_hscan->rs_cbuf);
		fc_xmax = HeapTupleHeaderGetRawXmax(fc_tuple->t_data);
		fc_infomask = fc_tuple->t_data->t_infomask;

		/*
		 * 如果 HTSU 返回 BeingModified，则一个元组是被锁定的。
		 */
		if (fc_htsu == TM_BeingModified)
		{
			fc_values[Atnum_tid] = (char *) DirectFunctionCall1(tidout,
															 PointerGetDatum(&fc_tuple->t_self));

			fc_values[Atnum_xmax] = palloc(NCHARS * sizeof(char));
			snprintf(fc_values[Atnum_xmax], NCHARS, "%u", fc_xmax);
			if (fc_infomask & HEAP_XMAX_IS_MULTI)
			{
				MultiXactMember *fc_members;
				int			fc_nmembers;
				bool		fc_first = true;
				bool		fc_allow_old;

				fc_values[Atnum_ismulti] = pstrdup("true");

				fc_allow_old = HEAP_LOCKED_UPGRADED(fc_infomask);
				fc_nmembers = GetMultiXactIdMembers(fc_xmax, &fc_members, fc_allow_old,
												 false);
				if (fc_nmembers == -1)
				{
					fc_values[Atnum_xids] = "{0}";
					fc_values[Atnum_modes] = "{transient upgrade status}";
					fc_values[Atnum_pids] = "{0}";
				}
				else
				{
					int			fc_j;

					fc_values[Atnum_xids] = palloc(NCHARS * fc_nmembers);
					fc_values[Atnum_modes] = palloc(NCHARS * fc_nmembers);
					fc_values[Atnum_pids] = palloc(NCHARS * fc_nmembers);

					strcpy(fc_values[Atnum_xids], "{");
					strcpy(fc_values[Atnum_modes], "{");
					strcpy(fc_values[Atnum_pids], "{");

					for (fc_j = 0; fc_j < fc_nmembers; fc_j++)
					{
						char		fc_buf[NCHARS];

						if (!fc_first)
						{
							strcat(fc_values[Atnum_xids], ",");
							strcat(fc_values[Atnum_modes], ",");
							strcat(fc_values[Atnum_pids], ",");
						}
						snprintf(fc_buf, NCHARS, "%u", fc_members[fc_j].xid);
						strcat(fc_values[Atnum_xids], fc_buf);
						switch (fc_members[fc_j].status)
						{
							case MultiXactStatusUpdate:
								snprintf(fc_buf, NCHARS, "Update");
								break;
							case MultiXactStatusNoKeyUpdate:
								snprintf(fc_buf, NCHARS, "No Key Update");
								break;
							case MultiXactStatusForUpdate:
								snprintf(fc_buf, NCHARS, "For Update");
								break;
							case MultiXactStatusForNoKeyUpdate:
								snprintf(fc_buf, NCHARS, "For No Key Update");
								break;
							case MultiXactStatusForShare:
								snprintf(fc_buf, NCHARS, "Share");
								break;
							case MultiXactStatusForKeyShare:
								snprintf(fc_buf, NCHARS, "Key Share");
								break;
						}
						strcat(fc_values[Atnum_modes], fc_buf);
						snprintf(fc_buf, NCHARS, "%d",
								 BackendXidGetPid(fc_members[fc_j].xid));
						strcat(fc_values[Atnum_pids], fc_buf);

						fc_first = false;
					}

					strcat(fc_values[Atnum_xids], "}");
					strcat(fc_values[Atnum_modes], "}");
					strcat(fc_values[Atnum_pids], "}");
				}
			}
			else
			{
				fc_values[Atnum_ismulti] = pstrdup("false");

				fc_values[Atnum_xids] = palloc(NCHARS * sizeof(char));
				snprintf(fc_values[Atnum_xids], NCHARS, "{%u}", fc_xmax);

				fc_values[Atnum_modes] = palloc(NCHARS);
				if (fc_infomask & HEAP_XMAX_LOCK_ONLY)
				{
					if (HEAP_XMAX_IS_SHR_LOCKED(fc_infomask))
						snprintf(fc_values[Atnum_modes], NCHARS, "{For Share}");
					else if (HEAP_XMAX_IS_KEYSHR_LOCKED(fc_infomask))
						snprintf(fc_values[Atnum_modes], NCHARS, "{For Key Share}");
					else if (HEAP_XMAX_IS_EXCL_LOCKED(fc_infomask))
					{
						if (fc_tuple->t_data->t_infomask2 & HEAP_KEYS_UPDATED)
							snprintf(fc_values[Atnum_modes], NCHARS, "{For Update}");
						else
							snprintf(fc_values[Atnum_modes], NCHARS, "{For No Key Update}");
					}
					else
						/* 既没有设置 keyshare 也没有设置 exclusive 位 */
						snprintf(fc_values[Atnum_modes], NCHARS,
								 "{transient upgrade status}");
				}
				else
				{
					if (fc_tuple->t_data->t_infomask2 & HEAP_KEYS_UPDATED)
						snprintf(fc_values[Atnum_modes], NCHARS, "{Update}");
					else
						snprintf(fc_values[Atnum_modes], NCHARS, "{No Key Update}");
				}

				fc_values[Atnum_pids] = palloc(NCHARS * sizeof(char));
				snprintf(fc_values[Atnum_pids], NCHARS, "{%d}",
						 BackendXidGetPid(fc_xmax));
			}

			LockBuffer(fc_hscan->rs_cbuf, BUFFER_LOCK_UNLOCK);

			/* 构建一个元组 */ 
			fc_tuple = BuildTupleFromCStrings(fc_attinmeta, fc_values);
			tuplestore_puttuple(fc_rsinfo->setResult, fc_tuple);
		}
		else
		{
			LockBuffer(fc_hscan->rs_cbuf, BUFFER_LOCK_UNLOCK);
		}
	}

	table_endscan(fc_scan);
	table_close(fc_rel, AccessShareLock);
	return (Datum) 0;
}
